package listener

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/sirupsen/logrus"
	"miao-server/global"
	"miao-server/lib/network"
	"miao-server/structs/base"
	"miao-server/structs/dto"
	"miao-server/structs/enum"
	"net"
	"runtime"
	"sync"
	"time"
)

///////////////////// TcpServer 代码  /////////////////////////////////

type TcpClient struct {
	ClientId   int64
	UniqueCode string
	tcp        net.Conn
	inChan     chan base.ClientResponse
	outChan    chan base.ClientInstruct
	closeChan  chan byte
	mutex      sync.Mutex
	IsClosed   bool
	tcpServer  *tcpServer
}

var TcpSer *tcpServer

type tcpServer struct {
	server      string
	clientMap   map[string]*TcpClient
	clientMutex sync.Mutex
}

func NewTcpServer(engine string) *tcpServer {
	server := &tcpServer{
		server:      engine,
		clientMap:   make(map[string]*TcpClient),
		clientMutex: sync.Mutex{},
	}
	return server
}

func (s *tcpServer) NewClient(conn net.Conn) *TcpClient {
	client := &TcpClient{
		ClientId:   0,
		UniqueCode: "",
		tcp:        conn,
		inChan:     make(chan base.ClientResponse, 1000),
		outChan:    make(chan base.ClientInstruct, 1000),
		closeChan:  make(chan byte, 1),
		mutex:      sync.Mutex{},
		IsClosed:   false,
		tcpServer:  s,
	}
	// 启动读信息的协程
	go client.readLoop()
	// 启动写协程
	go client.writeLoop()
	return client
}
func InitTcpServer() {
	TcpSer = NewTcpServer(global.Conf.Tcp.Server)
	go func() {
		err := TcpSer.Start()
		if err != nil {
			logrus.Errorf("启动tcp服务失败:%s", err)
			return
		}
	}()
	go func() {
		ticker := time.NewTicker(time.Duration(global.Conf.Tcp.Interval) * time.Second)
		for range ticker.C {
			for _, v := range TcpSer.clientMap {
				msg := base.ClientInstruct{
					Code: enum.Heartbeat,
				}
				err := v.WriteMessage(msg)
				if err != nil {
					logrus.Debugf("发送消息：err :%v", err)
					TcpSer.clientOffline(v)
				}
			}
		}
	}()

}
func (s *tcpServer) Start() error {
	listener, err := net.Listen("tcp", s.server)
	defer listener.Close()
	if err != nil {
		logrus.Errorf("启动tcp监听：%s ,失败：%v", s.server, err)
		return err
	}
	for {
		con, err := listener.Accept()
		if err != nil {
			logrus.Errorf("客户端连接失败：%v", err)
			continue
		}
		client := s.NewClient(con)
		logrus.Debugf("客户上线：%+v", client)
		// 开启协程，监听客户端消息
		go client.handMsg()
	}
}

// IsOnline 判断终端是否在线
func (s *tcpServer) IsOnline(uniqueCode string) (b bool) {
	_, b = s.clientMap[uniqueCode]
	return b
}

// SendInstruct 给终端发送消息
func (s *tcpServer) SendInstruct(uniqueCode string, in base.ClientInstruct) (err error) {
	if !s.IsOnline(uniqueCode) {
		return errors.New("设备不在线")
	}
	err = s.clientMap[uniqueCode].WriteMessage(in)
	return err
}

func (s *tcpServer) GetClientInfo() (li []dto.TcpClientDto) {
	for _, v := range s.clientMap {
		li = append(li, dto.TcpClientDto{
			ClientId:   v.ClientId,
			UniqueCode: v.UniqueCode,
			HandsUpMsg: global.HandUpCache[v.UniqueCode],
		})
	}
	return
}

func (c *TcpClient) handMsg() {
	for {
		in, err := c.ReaderMessage()
		if err != nil {
			break
		}
		// 客户端每关闭连接才处理消息，有两个原因：
		// 1. 尽量确保ACK出错的概率，通道关闭就不能做二次确认了
		// 2. channel 是在协程中读取的，未关闭时是阻塞状态，关闭后会读取到一条零值，这样可以规避那个零值信息
		if !c.IsClosed {
			switch in.ContentCode {
			case enum.MsgForHeartbeat:
				if c.UniqueCode == "" {
					inst := base.ClientInstruct{
						Code: enum.HandsUp,
					}
					err = c.WriteMessage(inst)
				} else {
					c.HeartBeat(in)
				}
			case enum.MsgForGetDeviceId:
				err = c.GetDeviceId(in)
				if err != nil {
					logrus.Error(err)
				}
			case enum.MsgForHandsUp:
				var msg dto.HandsUpMsg
				err = json.Unmarshal([]byte(in.Result), &msg)
				if err != nil {
					continue
				}
				var did int64
				did, err = c.HandUp(msg)
				if err != nil {
					continue
				}
				if did != 0 {
					// 说明是正常设备
					c.ClientId = did
					inst := base.ClientInstruct{
						Code: enum.SetDeviceId,
						Body: fmt.Sprintf("%d", did),
					}
					err = c.WriteMessage(inst)
					if err != nil {
						logrus.Errorf("发送数据时发生错误")
					}
				}
			case enum.MsgForUpdateAllPolicy:
				err = c.sendClientPolicy()
				if err != nil {
					logrus.Error(err)
				}
			case enum.MsgForExecutePlugin:
				c.cacheResponse(in, enum.ExecutePlugin)
			case enum.MsgForUploadPolicy:
				c.cacheResponse(in, enum.UploadPolicy)
			case enum.MsgForLoadCount:
				err = c.handPcLoad(in)
			default:
				logrus.Errorf("未知的消息类型：%+v", in)
			}
		}
	}
}
func (s *tcpServer) clientOffline(client *TcpClient) {
	// 移除 map中的连接记录
	defer s.clientMutex.Unlock()
	s.clientMutex.Lock()
	// 关闭连接，释放句柄
	client.Close()
	// 通知协程退出
	runtime.Goexit()
}

func (c *TcpClient) ReaderMessage() (msg base.ClientResponse, err error) {
	select {
	case msg = <-c.inChan:
	case <-c.closeChan:
		err = errors.New("connection is closed")
	}
	return
}

func (c *TcpClient) WriteMessage(msg base.ClientInstruct) (err error) {
	msg.ServeTime = time.Now().Unix()
	select {
	case c.outChan <- msg:
	case <-c.closeChan:
		err = errors.New("connection is closed")
	}
	return
}

func (c *TcpClient) readLoop() {
	var (
		bs  []byte
		err error
	)
	for {
		if !c.IsClosed {
			if bs, err = network.Decode(bufio.NewReader(c.tcp)); err != nil {
				goto ERR
			}
			inst := base.ClientResponse{}
			err = json.Unmarshal(bs, &inst)
			if err != nil {
				logrus.Errorf("读取到不合法的消息：%s", string(bs))
				goto ERR
			}
			select {
			case c.inChan <- inst:
			case <-c.closeChan:
				goto ERR
			}
		}
	}
	// 发生错误时应该退出循环，否则协程无法正常回收
ERR:
	c.Close()
}

func (c *TcpClient) writeLoop() {
	for {
		select {
		case msg := <-c.outChan:
			if !c.IsClosed {
				bs, err := network.Encode(msg)
				if err != nil {
					logrus.Errorf("tcp消息【%+v】编码错误,err:%v", msg, err)
					goto ERR
				}
				_, err = c.tcp.Write(bs)
				if err != nil {
					goto ERR
				}
			}
		case <-c.closeChan:
			goto ERR
		}

	}
ERR:
	c.Close()
}

func (c *TcpClient) Close() {
	defer c.mutex.Unlock()
	c.mutex.Lock()
	// 关闭操作只能执行一次
	if c.tcp != nil {
		_ = c.tcp.Close()
	}
	if !c.IsClosed {
		logrus.Debugf("客户端下线：%+v", c)
		close(c.closeChan)
		close(c.inChan)
		close(c.outChan)
		c.IsClosed = true
	}
	delete(c.tcpServer.clientMap, c.UniqueCode)
}
